package blue.mqtt4j.internal.client.impl;

import blue.mqtt4j.client.MqttClientException;
import blue.mqtt4j.client.MqttMessage;
import blue.mqtt4j.client.MqttProducerException;
import blue.mqtt4j.client.Producer;
import blue.mqtt4j.client.ProducerListener;
import blue.mqtt4j.client.ProducerResult;
import blue.mqtt4j.client.codec.Encoder;
import blue.mqtt4j.client.codec.JsonEncoder;
import blue.mqtt4j.core.util.AssertUtil;
import blue.mqtt4j.core.util.WaitUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author Jin Zheng
 * @since 1.0 2020-09-30
 */
public class DefaultProducer implements Producer
{
	private static Logger logger = LoggerFactory.getLogger(DefaultProducer.class);

	private final DefaultMqttClient mqttClient;
	private ProducerListener defaultListener = new LoggerProducerListener();
	private Encoder defaultEncoder = new JsonEncoder();
	private boolean connectState = true;

	public DefaultProducer(DefaultMqttClient mqttClient)
	{
		this.mqttClient = mqttClient;
	}

	@Override
	public void sendSync(List<MqttMessage> messageList, Encoder encoder)
	{
		this.checkState();
		AssertUtil.isEmpty(messageList, "Message list");
		Encoder localEncoder = encoder == null ? defaultEncoder : encoder;
		CountDownLatch latch = new CountDownLatch(messageList.size());
		int i = 0;
		BlockingQueue<ProducerResult> results = new LinkedBlockingQueue<>();
		for (MqttMessage message : messageList)
		{
			ProducerListener listener = new SyncProducerListener(latch, results, i);
			mqttClient.publish(message, localEncoder, listener, false);
			i++;
		}
		WaitUtil.await(latch, mqttClient.getOptions().timeoutMillis(), TimeUnit.MILLISECONDS);
		if (!results.isEmpty())
			throw new MqttProducerException(results);
	}

	@Override
	public void sendAsync(List<MqttMessage> messageList, Encoder encoder, ProducerListener listener)
	{
		AssertUtil.isEmpty(messageList, "Message list");
		Encoder localEncoder = encoder == null ? defaultEncoder : encoder;
		ProducerListener localListener = listener == null ? this.defaultListener : listener;
		for (MqttMessage message : messageList)
		{
			mqttClient.publish(message, localEncoder, localListener,  true);
		}
	}

	@Override
	public void setProducerListener(ProducerListener listener)
	{
		if (listener != null)
		{
			this.defaultListener = listener;
			logger.info("Set default producer listener: {}", listener.getClass().getName());
		}
	}

	@Override
	public void setEncoder(Encoder encoder)
	{
		if (encoder != null)
		{
			this.defaultEncoder = encoder;
			logger.info("Set default encoder: {}", defaultEncoder.getClass().getName());
		}
	}

	private void checkState()
	{
		if (!connectState)
			throw new MqttClientException("MqttClient was disconnected");
	}

	public void setConnectState(boolean state)
	{
		this.connectState = state;
	}
}
